package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.app.func.DimAsyncFunction;
import com.atguigu.edu.realtime.bean.AppVideoBean;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.MyClickhouseUtil;
import com.atguigu.edu.realtime.util.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

public class DwsAppVideoWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        String topic = "dwd_traffic_appVideo_log";
        String groupId = "dws_appVideo_window";
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        DataStreamSource<String> kafkaDS = env.addSource(kafkaConsumer);

        SingleOutputStreamOperator<AppVideoBean> mapDS = kafkaDS.map(
                new MapFunction<String, AppVideoBean>() {
                    @Override
                    public AppVideoBean map(String s) throws Exception {
                        JSONObject jsonObj = JSON.parseObject(s);
                        String user_id = jsonObj.getJSONObject("common").getString("uid");
                        String video_id = jsonObj.getJSONObject("appVideo").getString("video_id");
                        String during_time = jsonObj.getJSONObject("appVideo").getString("position_sec");
                        Long ts = jsonObj.getLong("ts");
                        AppVideoBean appVideoBean = new AppVideoBean(
                                "",
                                "",
                                user_id,
                                video_id,
                                "",
                                "",
                                during_time,
                                ts
                        );
                        return appVideoBean;
                    }
                }
        );

        SingleOutputStreamOperator<AppVideoBean> withWaterMarkDS = mapDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<AppVideoBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<AppVideoBean>() {
                                    @Override
                                    public long extractTimestamp(AppVideoBean appVideoBean, long l) {
                                        return appVideoBean.getTs();
                                    }
                                }
                        )
        );

        KeyedStream<AppVideoBean, String> keyedDS = withWaterMarkDS.keyBy(appVideoBean -> appVideoBean.getVideoId());

        WindowedStream<AppVideoBean, String, TimeWindow> windowDS = keyedDS.window(TumblingEventTimeWindows.of(Time.seconds(10)));

        SingleOutputStreamOperator<AppVideoBean> reduceDS = windowDS.reduce(
                new ReduceFunction<AppVideoBean>() {
                    @Override
                    public AppVideoBean reduce(AppVideoBean v1, AppVideoBean v2) throws Exception {
                        v1.setTimeSum(v1.getTimeSum() + v2.getTimeSum());
                        return v1;
                    }
                },
                new WindowFunction<AppVideoBean, AppVideoBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow timeWindow, Iterable<AppVideoBean> iterable, Collector<AppVideoBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(timeWindow.getStart());
                        String edt = DateFormatUtil.toYmdHms(timeWindow.getEnd());
                        for (AppVideoBean appVideoBean : iterable) {
                            appVideoBean.setStt(stt);
                            appVideoBean.setEdt(edt);
                            appVideoBean.setTs(System.currentTimeMillis());
                            out.collect(appVideoBean);
                        }
                    }
                }
        );
        SingleOutputStreamOperator<AppVideoBean> withChapterIdDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<AppVideoBean>("dim_video_info") {
                    @Override
                    public void join(AppVideoBean obj, JSONObject dimInfoJsonObj) {
                        obj.setChapterId(dimInfoJsonObj.getString("CHAPTER_ID"));
                    }

                    @Override
                    public String getKey(AppVideoBean obj) {
                        return obj.getVideoId();
                    }
                }, 60, TimeUnit.SECONDS
        );
        SingleOutputStreamOperator<AppVideoBean> withChapterNameDS = AsyncDataStream.unorderedWait(
                withChapterIdDS,
                new DimAsyncFunction<AppVideoBean>("dim_chapter_info") {
                    @Override
                    public void join(AppVideoBean obj, JSONObject dimInfoJsonObj) {
                        obj.setChapterName(dimInfoJsonObj.getString("CHAPTER_NAME"));
                    }

                    @Override
                    public String getKey(AppVideoBean obj) {
                        return obj.getChapterId();
                    }
                }, 60, TimeUnit.SECONDS
        );

        withChapterNameDS.print(">>>>");
        withChapterNameDS.addSink(MyClickhouseUtil.<AppVideoBean>getSinkFunction("insert into dws_appVideo_view_window values(?,?,?,?,?,?,?,?)"));


        env.execute();

    }
}
